package com.github.llyb120.qiuya.storage.bptree;

import com.github.llyb120.qiuya.error.NotInTransactionException;
import com.github.llyb120.qiuya.storage.VirtualFile;

import java.io.File;
import java.util.*;

import static com.github.llyb120.qiuya.storage.bptree.Node.*;

public final class BPlusTree {

    //源文件,并不实际创建
//    private File sourceFile;
    //块大小
//    private int blockSize;
    //实际操作的抽象文件
    private VirtualFile file;

    //根节点偏移
    private long root;

    Set<Node> modified = new HashSet<>();
    Map<Long, Node> nodeCache = new HashMap<>();
    Map<Long, Long> parentCache = new HashMap<>();
    //可用的块
    TreeMap<Long, Boolean> freeBlocks = new TreeMap<>();

    public static long INVALID_OFFSET = 0xdeadbeef;
    private long currentOffset = 0L;

    private volatile boolean inTransaction = false;

    public long getNewOffset(){
        //优先从已删除的块中查找
        if(!freeBlocks.isEmpty()){
            Long key = freeBlocks.firstKey();
            freeBlocks.remove(key);
            return key;
        }
        return getCurrentOffsetAndStep();
    }

    long getCurrentOffsetAndStep() {
        long ret = currentOffset;
        currentOffset += BLOCK_SIZE;
        return ret;
    }

//    public BPlusTree(File file) {
//        this.sourceFile = file;
//        init();
//    }

    public BPlusTree(VirtualFile vf, long root){
        file = vf;
        this.root = root;
    }

//    void init() {
//        root = INVALID_OFFSET;
//        file = new VirtualFile(sourceFile);
//        开始读取文件
//    }

    void recordParent(long child, long parent) {
        parentCache.put(child, parent);
    }

    long getParent(Node node) {
        Long item = parentCache.get(node.self);
        if (item == null) {
            return INVALID_OFFSET;
        }
        return item;
    }

    Node createKeyNode(long parent) {
        Node node = new Node();
        node.self = getNewOffset();//getCurrentOffsetAndStep();
        recordParent(node.self, parent);
        node.isLeaf = false;

        nodeCache.put(node.self, node);
        return node;
    }

    Node createDataNode(long parent) {
        Node node = new Node();
        node.self = getNewOffset();//getCurrentOffsetAndStep();
        recordParent(node.self, parent);
        node.isLeaf = true;

        nodeCache.put(node.self, node);
        return node;
    }


    int searchOffset(Node node, long key) {
        NodeData last = null;
        int i = 0;
        for (NodeData datum : node.data) {
            if (last != null) {
                if (key > last.key && key <= datum.key) {
                    return i;
                }
            }
            last = datum;
            i++;
        }
        return -1;
    }

    public boolean begin() {
        if (inTransaction) {
            return false;
        }
        clearCache();
        inTransaction = true;
        return true;
    }

    /**
     * 变更提交
     */
    public void commit() {
        if (!inTransaction) {
            return;
        }
        try{
            //写入磁盘
            for (Node node : modified) {
                flushNode(node);
            }
        } finally {
            clearCache();
            inTransaction = false;
        }
    }

    /**
     * 将节点写回磁盘
     * @param node
     */
    void flushNode(Node node){

    }


    public Node firstLeafNode(){
        Node node = seek(root);
        if (node == null) {
            return null;
        }
        while (node != null) {
            if (node.isLeaf) {
                return node;
            } else {
                if(node.data.isEmpty()){
                    return null;
                }
                node = seek(node.data.get(0).value);
            }
        }
        return null;
    }

    void clearCache(){
        modified.clear();
        nodeCache.clear();
        parentCache.clear();
    }

    @Deprecated
    public void rollback(){
        if(!inTransaction){
            return;
        }
        clearCache();
    }

    public void put(long key, long value) {
        System.out.printf("puting %d => %d\n", key, value);
        boolean tran = begin();
        try {
            Node node = seek(root);
            if (node == null) {
                node = createDataNode(root);
                root = node.self;
            }
            doOnDataNode(node, key, (n) -> {
                replaceByOffset(n, true, key, key, value);
                return null;
            }, null);
        } finally {
            if(tran){
                commit();
            }
        }
    }

    public long get(long key) {
        boolean tran = begin();
        try {
            Node node = seek(root);
            return (long) doOnDataNode(node, key, (n) -> {
                return n.data.stream()
                        .filter(e -> e.key == key)
                        .map(e -> e.value)
                        .findFirst()
                        .orElse(INVALID_OFFSET);
            }, INVALID_OFFSET);
        } finally {
            if(tran){
                commit();
            }
        }
    }

    public void remove(long key) {
        boolean tran = begin();
        try {
            Node node = seek(root);
            if (node == null) {
                return;
            }
            doOnDataNode(node, key, (n) -> {
                remove(n, true, key);
                return null;
            }, null);
        } finally {
            if(tran){
                commit();
            }
        }
    }

    Object doOnDataNode(Node node, long key, LeafNodeAction action, Object defaultValue) {
        while (node != null) {
            if (node.isLeaf) {
                Object ret = action.call(node);
                if (ret == null) {
                    return defaultValue;
                }
                return ret;
            } else {
                int pos = searchOffset(node, key);
                if (pos == -1) {
                    pos = node.data.size() - 1;
                }
                node = seek((node).data.get(pos).value);
            }
        }
        return defaultValue;
    }

    void remove(Node node, boolean useKey, long key) {
        if (node == null) {
            return;
        }
        node.data.removeIf(e -> useKey ? e.key == key : e.value == key);
        //如果节点为空
        if (node.data.isEmpty()) {
            //删除该节点
            Node left = seek(node.prev);
            if (left != null) {
                left.next = node.next;
                modified.add(left);
            }
            Node right = seek(node.next);
            if (right != null) {
                right.prev = node.prev;
                modified.add(right);
            }

            //设置空区
            freeBlocks.put(node.self, Boolean.TRUE);
            //清空改动记录
            modified.remove(node);
            //清空缓存，防止有新使用该记录的
            nodeCache.remove(node.self);

            remove(seek(getParent(node)), false, node.self);
        } else {
            modified.add(node);
//            replaceByOffset(seek(getParent(node)), false, node.self, node.getLastData().key, node.self);
        }
    }


    void splitNode(Node left) {
        Node right = null;
        if (left.isLeaf) {
            right = createDataNode(getParent(left));
        } else {
            right = createKeyNode(getParent(left));
        }
        left.next = right.self;
        right.prev = left.self;
        int len = left.data.size();
        int mid = len / 2;
        List<NodeData> source = left.data;
        right.data = new ArrayList<>(source.subList(mid, len));
        left.data = new ArrayList<>(source.subList(0, mid));

        modified.add(left);
        modified.add(right);

        doAfterSplit(left, right, mid);
    }

    void replaceByOffset(Node node, boolean leafNodePutVal, long... items) {
        if (node == null) {
            return;
        }
        node.data.removeIf(e -> leafNodePutVal ? e.key == items[0] : e.value == items[0]);
        for (int i = 1; i < items.length; i += 2) {
            (node).data.add(new NodeData(items[i], items[i + 1]));
        }
        Collections.sort(node.data, Comparator.comparingLong(a -> a.key));
        modified.add(node);
        if (node.canSplit()) {
            splitNode(node);
        } else {
//            replaceByOffset(seek(getParent(node)), false, node.self, node.getLastData().key, node.self);
        }
    }



    void doAfterSplit(Node left, Node right, int mid) {
        long parent = getParent(left);
        Node parNode = seek(parent);
        if (parNode == null) {
            parNode = createKeyNode(parent);//tree.getCurrentOffsetAndStep());
            replaceByOffset(parNode, false, -1, left.getLastData().key, left.self);
            replaceByOffset(parNode, false, -1, right.getLastData().key, right.self);
        } else {
            replaceByOffset(parNode, false, left.self, left.getLastData().key, left.self, right.getLastData().key, right.self);
        }

        //修复该节点
        fixNodeChildren(left);
        fixNodeChildren(right);
        fixNodeChildren(parNode);

        if (left.self == root) {
            root = parNode.self;
        }
//        maxKeyToParent(left, right);
        modified.add(parNode);
    }

    void fixNodeChildren(Node node) {
        if (node.isLeaf) {
            return;
        }
        for (NodeData datum : node.data) {
            recordParent(datum.value, node.self);
        }
    }


    public Node seek(long offset) {
        if (offset == INVALID_OFFSET) {
            return null;
        }
        if(freeBlocks.containsKey(offset)){
            return null;
        }
        return nodeCache.get(offset);
    }


    public void dump() {
        //得到root
        Node rootNode = nodeCache.values().stream()
                .filter(e -> e.self == root)
                .findFirst()
                .orElse(null);


        dump(rootNode, 0);

        System.out.println("***************");
    }

    void dump(Node node, int offset) {
        if (node == null) {
            return;
        }
        if (!node.isLeaf) {
            dumpKeyNode(node, offset);
        } else {
            dumpDataNode(node, offset);
        }
    }

    void dumpOffset(int offset) {
        for (int i = 0; i < offset; i++) {
            System.out.printf(" ");
        }
    }

    void dumpKeyNode(Node node, int offset) {
        dumpOffset(offset);
        System.out.printf("[%d:", node.self);
        for (NodeData datum : node.data) {
            System.out.printf(" %d ", datum.key);
        }
        System.out.printf("]\n");
        for (NodeData datum : node.data) {
            Node child = nodeCache.get(datum.value);
            dump(child, offset + 4);
        }
    }

    void dumpDataNode(Node node, int offset) {
        dumpOffset(offset);
        System.out.printf("{%d: ", node.self);
        for (NodeData datum : node.data) {
            System.out.printf(" %d => %d", datum.key, datum.value);
        }
        System.out.printf("}\n");
    }


}
